package com.my.redisson.delayqueue;

import com.my.redisson.delayqueue.handler.DelayEventHandlerDispatcher;
import com.my.redisson.domain.DelayEvent;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
@Component
public class DelayQueueConsumer {

    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private ThreadPoolExecutor executor;
    @Autowired
    private DelayEventHandlerDispatcher dispatcher;

    @Value("delay_queue_call")
    private String group;

    private DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss:SSS");

    @PostConstruct
    public void start() {
        executor.execute(() -> {
            RBlockingQueue<DelayEvent> blockingFairQueue = redissonClient.getBlockingQueue(group);
            // 开启客户端监听（必须调用），否者系统重启时拿不到已过期数据，要等到系统第一次调用getDelayedQueue方法时才能开启监听
            redissonClient.getDelayedQueue(blockingFairQueue);
            log.info("DelayQueueConsumer started group:{}", group);
            while (true) {
                DelayEvent event;
                try {
                    event = blockingFairQueue.take();
                } catch (Exception e) {
                    continue;
                }
                if (Objects.isNull(event)) {
                    continue;
                }
                LocalDateTime localDateTime = Instant.ofEpochMilli(event.getOnTime()).atOffset(ZoneOffset.ofHours(8)).toLocalDateTime();
                log.info("DelayQueueConsumer receive=======event:{} group:{} onTimeFmt:{}", event, group, formatter.format(localDateTime));
                dispatcher.handle(event);
            }
        });
    }

}
